package studio.raptor.dispatcher.server.listeners;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import studio.raptor.dispatcher.scheduler.ServerScheduler;
import studio.raptor.dispatcher.server.dto.Consumer;
import studio.raptor.dispatcher.server.event.ConsumerCreationEvent;
import studio.raptor.dispatcher.server.event.ConsumerUpdatedEvent;
import studio.raptor.dispatcher.server.task.ConsumerTask;
import studio.raptor.dispatcher.server.task.DispatcherTask;
import studio.raptor.hub.out.HubOut;

/**
 * @author jack
 * @since 1.0
 */
@Component
public class ConsumerEventListener {

  private static final Logger log = LoggerFactory.getLogger(ConsumerEventListener.class);

  private final Map<String, ConsumerTask> consumerTaskMap = new ConcurrentHashMap<>(100);
  private final Map<String, DispatcherTask> dispatcherTaskMap = new ConcurrentHashMap<>(100);

  @Autowired
  HubOut hubOut;

  @Autowired
  ApplicationContext context;

  @Autowired
  ThreadPoolTaskExecutor threadPoolTaskExecutor;

  @Autowired
  ServerScheduler serverScheduler;

  @Async
  @EventListener
  public void handleConsumerCreatedEvent(ConsumerCreationEvent creationEvent) {
    Consumer consumer = creationEvent.getConsumer();

    String key = consumer.getCode() + "-" + consumer.getKindId();

    if (consumerTaskMap.containsKey(key)) {
      log.warn("consumer：{}-{} already exists", consumer.getCode(), consumer.getName());
      return;
    }

    boolean canHandle = hubOut.getCodes().contains(key);
    if (!canHandle) {
      log.warn("the task[{}] can't be processed", key);
      return;
    }

    log.info("add consumer->{}", consumer.toString());
    ConsumerTask consumerTask = createConsumerTask(consumer);
    threadPoolTaskExecutor.submit(consumerTask);
    consumerTaskMap.put(key, consumerTask);

    if(consumer.isSort()) {
      DispatcherTask dispatcherTask = createDispatcherTask(consumer);
      threadPoolTaskExecutor.submit(dispatcherTask);
      dispatcherTaskMap.put(key, dispatcherTask);
    }

  }

  @Async
  @EventListener
  public void handleConsumerUpdateEvent(ConsumerUpdatedEvent updatedEvent) {
    Consumer consumer = updatedEvent.getConsumer();
    String key = consumer.getCode() + "-" + consumer.getKindId();
    log.info("update consumer [{}] state, new state is:{}", key,
        consumer.isState() ? "start" : "stop");
    if (consumerTaskMap.containsKey(key)) {
      log.info("update consumer state:{}", consumer.toString());
      ConsumerTask consumerTask = consumerTaskMap.get(key);
      DispatcherTask dispatcherTask = dispatcherTaskMap.get(key);
      consumerTask.setStart(consumer.isState());
      if (consumer.isSort()) {
        dispatcherTask.setStart(consumer.isState());
      }
    }

  }

  public Map<String, ConsumerTask> getConsumerTaskMap() {
    return consumerTaskMap;
  }

  public Map<String, DispatcherTask> getDispatcherTaskMap() {
    return dispatcherTaskMap;
  }

  private ConsumerTask createConsumerTask(Consumer consumer) {
    ConsumerTask consumerTask = context.getBean(ConsumerTask.class);
    consumerTask.setConsumer(consumer);
    consumerTask.setStart(consumer.isState());
    return consumerTask;
  }

  private DispatcherTask createDispatcherTask(Consumer consumer) {
    DispatcherTask dispatcherTask = context.getBean(DispatcherTask.class);
    dispatcherTask.setConsumer(consumer);
    dispatcherTask.setStart(consumer.isState());
    return dispatcherTask;
  }


}
